/**
 * 
 */
package com.rrd.dw.mr.all;

import java.io.IOException;
// import java.util.HashMap;
// import java.util.Map;
import java.io.InputStream;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.amazonaws.util.Md5Utils;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HBaseUtil;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.MD5Util;

/**
 * @author liubaoxin
 */
public class PhoneBookRecordsExtractor extends Configured implements Tool {
	private static final String STR_FILED_SPLIT_FLAG = "\001";

	public static final int SIZEOF_MD5 = 16;

	private static final byte[] FAMILY_INF = Bytes.toBytes("inf");
	private static final byte[] FAMILY_JOB = Bytes.toBytes("job");
	// private static final byte[] COLUMN_INF_NAME = Bytes.toBytes("name");
	// private static final byte[] COLUMN_INF_MOBILE = Bytes.toBytes("mobile");
	// private static final byte[] COLUMN_INF_REMARK = Bytes.toBytes("remark");

	private static final String STR_TARGET_POSITION_PHONE_BOOK_RECORDS_INF = "phonebookrecordsbasic";
	private static final String STR_TARGET_POSITION_PHONE_BOOK_RECORDS_JOB_INF = "phonebookrecordsjob";

	private static class PhoneBookRecordsMapper extends TableMapper<Text, Text> {
		private NullWritable key2 = NullWritable.get();
		private Text key = null;
		private Text value = null;
		private MultipleOutputs<Text, Text> mos = null;
		protected void setup(Context context) {

			key = new Text();
			value = new Text();
			mos = new MultipleOutputs<Text, Text>(context);
		}

		protected void cleanup(Context context) throws IOException, InterruptedException {
			mos.close();
		}


		protected void map(ImmutableBytesWritable key, Result value, Context context)
				throws IOException, InterruptedException {
			String keyStr = null;
			StringBuffer keysb = new StringBuffer();
//			Thread.sleep(5);
			Object[] objs = HBaseUtil.decodeUserRowkey(key.get());
			if (objs == null)
				return;

			String userKey = objs[0].toString();
			long timestamp = Long.parseLong(objs[1].toString());

			keysb.append(userKey);
			keysb.append(timestamp);
			keysb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("job_id"))));
			keysb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("system_id")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("system_id"))));
			keysb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")) == null ? ""
					: Bytes.toLong(value.getValue(FAMILY_JOB, Bytes.toBytes("end_time"))));
			keyStr = MD5Util.getMd5(keysb.toString());
			this.key.set(STR_TARGET_POSITION_PHONE_BOOK_RECORDS_JOB_INF);

			StringBuffer sb = new StringBuffer();
			sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
			sb.append(userKey).append(STR_FILED_SPLIT_FLAG);
			sb.append(timestamp).append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")))).append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("finished")) == null ? ""
					: String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB, Bytes.toBytes("finished")))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("requested")) == null ? ""
					: String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB, Bytes.toBytes("requested")))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("successful")) == null ? ""
					: String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB, Bytes.toBytes("successful")))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("start_time")) == null ? ""
					: String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB, Bytes.toBytes("start_time")))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")) == null ? ""
					: String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("data_file")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("data_file"))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("system_id")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("system_id"))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("query")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("query")))).append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("try_times")) == null ? ""
					: String.valueOf(Bytes.toInt(value.getValue(FAMILY_JOB, Bytes.toBytes("try_times")))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("type")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("type")))).append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("message")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("message"))))
					.append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("code")) == null ? ""
					: Bytes.toString(value.getValue(FAMILY_JOB, Bytes.toBytes("code")))).append(STR_FILED_SPLIT_FLAG);
			sb.append(value.getValue(FAMILY_JOB, Bytes.toBytes("timestamp")) == null ? ""
					: String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB, Bytes.toBytes("timestamp")))));

			this.value.set(sb.toString());
			context.write(this.key, this.value);
//			this.mos.write(
//                    STR_TARGET_POSITION_PHONE_BOOK_RECORDS_JOB_INF, key2,
//                    value);
//			generatePhoneBookRecordsInfo(context, keyStr,  timestamp, value);
		}

		private void generatePhoneBookRecordsInfo(Context context, String keyStr, long timestamp,
				Result value) throws IOException, InterruptedException {
			StringBuffer sb = new StringBuffer();
			Cell countCell = null;
			int count = 0;

			countCell = value.getColumnLatestCell(FAMILY_INF, Bytes.toBytes("count"));
			if (countCell != null) {
				count = Bytes.toInt(CellUtil.cloneValue(countCell));

				for (int i = 0; i < count; i++) {
					sb.delete(0, sb.length());
					sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
					sb.append(timestamp).append(STR_FILED_SPLIT_FLAG);
					sb.append(i).append(STR_FILED_SPLIT_FLAG);

					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("name#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("name#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("mobile#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("mobile#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("remark#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("remark#" + i))))
							.append(STR_FILED_SPLIT_FLAG);

					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("email#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("email#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("title#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("title#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("phone_system#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("phone_system#" + i))))
							.append(STR_FILED_SPLIT_FLAG);

					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("phone_type#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("phone_type#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("call_count#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("call_count#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("called_time#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("called_time#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("create_time#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("create_time#" + i))))
							.append(STR_FILED_SPLIT_FLAG);
					sb.append(value.getValue(FAMILY_INF, Bytes.toBytes("modify_time#" + i)) == null ? ""
							: Bytes.toString(value.getValue(FAMILY_INF, Bytes.toBytes("modify_time#" + i))));

					String val = sb.toString().replace("\n", " ");
					val = MD5Util.getMd5(sb.toString()) + STR_FILED_SPLIT_FLAG + val;
					this.value.set(val);
					this.key.set(STR_TARGET_POSITION_PHONE_BOOK_RECORDS_INF);
					context.write(this.key, this.value);
//					 this.mos.write(STR_TARGET_POSITION_PHONE_BOOK_RECORDS_INF,
//		                        key2, value);
					 context.getCounter("phonebook","detail").increment(1);
				}
			}
		}
	}

	private static class PhoneBookRecordsReducer extends Reducer<Text, Text, Text, Text> {
		private NullWritable key2 = NullWritable.get();
		private MultipleOutputs<Text, Text> mos = null;

		protected void setup(Context context) throws IOException, InterruptedException {
			mos = new MultipleOutputs<Text, Text>(context);
		}

		protected void cleanup(Context context) throws IOException, InterruptedException {
			mos.close();
		}

		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			String keyStr = null;

			for (Text value : values) {
				keyStr = key.toString();

				if (STR_TARGET_POSITION_PHONE_BOOK_RECORDS_INF.equals(keyStr)) {
					this.mos.write(STR_TARGET_POSITION_PHONE_BOOK_RECORDS_INF, key2, value);
				} else if (STR_TARGET_POSITION_PHONE_BOOK_RECORDS_JOB_INF.equals(keyStr)) {
					this.mos.write(STR_TARGET_POSITION_PHONE_BOOK_RECORDS_JOB_INF, key2, value);
				}
			}
		}
	}

	@Override
	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub
		Properties prop = new Properties();
		try (InputStream is = this.getClass().getClassLoader().getResourceAsStream("hbase.properties");) {
			prop.load(is);
		}
		Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
		Configuration conf = job.getConfiguration();
		conf.set("hbase.zookeeper.quorum", "datanode1,datanode2,datanode3");
		conf.set("mapreduce.map.memory.mb", "3072");
		conf.set("mapreduce.reduce.memory.mb", "5120");
		conf.set("hbase.client.scanner.timeout.period", "800000");
//		conf.set("mapreduce.output.fileoutputformat.compress", "true");
//		conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
//		conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
		conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");

		/*
		 * conf.set("param.out.path",
		 * "hdfs://10.10.15.10:8020/tmp/output_xxh_rongreport/");
		 * conf.set("param.start_date", "2017-09-19");
		 * conf.set("param.stop_date", "2017-09-20");
		 */
		conf.set("param.out.path", args[0]);
		conf.set("param.start_date", args[1]);
		conf.set("param.stop_date", args[2]);
		String runMode = args[3];
		String tableName = "phone_book_records";
		String outputDir = conf.get("param.out.path");
		System.out.println("开始日期:" + conf.get("param.start_date"));
		System.out.println("结束日期:" + conf.get("param.stop_date"));
		System.out.println("输入HBase表名:" + tableName);
		System.out.println("输出路径:" + outputDir);

		Long startTime = DateUtil.formateToTimestamp(conf.get("param.start_date"));
		Long endTime = DateUtil.formateToTimestamp(conf.get("param.stop_date"));

		Path outpath = new Path(outputDir);
		FileSystem fs = outpath.getFileSystem(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
	
		job.setJarByClass(this.getClass());
		Scan scan = new Scan();

		scan.addFamily(FAMILY_INF);
		// scan.addColumn(FAMILY_INF, COLUMN_INF_NAME);
		// scan.addColumn(FAMILY_INF, COLUMN_INF_MOBILE);
		// scan.addColumn(FAMILY_INF, COLUMN_INF_REMARK);
		scan.addFamily(FAMILY_JOB);

		// scan.setCacheBlocks(false);
		// scan.setCaching(200);
		// scan.setBatch(100);
		scan.setMaxVersions(1);
		// scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME,
		// Bytes.toBytes(tableName));
		if (runMode == null || !runMode.equals("ALL")) {
			scan.setTimeRange(startTime, endTime);
			System.out.println("进行增量同步！！！！！！！！！");
		} else {
			System.out.println("进行全量同步！！！！！！！！！");
		}

		TableMapReduceUtil.initTableMapperJob(tableName, scan, PhoneBookRecordsMapper.class, Text.class, Text.class,
				job);
		FileOutputFormat.setOutputPath(job, outpath);
		MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_PHONE_BOOK_RECORDS_INF, TextOutputFormat.class,
				Text.class, Text.class);
		MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_PHONE_BOOK_RECORDS_JOB_INF, TextOutputFormat.class,
				Text.class, Text.class);

		 job.setReducerClass(PhoneBookRecordsReducer.class);
		job.setNumReduceTasks(50);

		int stat =job.waitForCompletion(true) ? 0 : 1;
		System.out.println("成功次數："+job.getCounters().getGroup("phonebook").findCounter("detail").getValue());
		return stat;
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		System.exit(ToolRunner.run(HBaseConfiguration.create(), new PhoneBookRecordsExtractor(), args));
	}
}
